<?php
declare (strict_types = 1);

namespace app\command;

use myshop\common\dao\AdminApiLogDao;
use think\console\Input;
use think\console\input\Argument;
use think\console\Output;

/**
 * 消费者服务
 */
class SimpleMQConsumer extends SimpleMQBase
{
    protected function configure()
    {
        // 指令配置
        $this->setName('SimpleMQConsumer')
            ->addArgument('name', Argument::OPTIONAL, "your name")
            ->setDescription('这是MQ简单的消费者工具');
    }

    protected function execute(Input $input, Output $output)
    {
        $name = $input->getArgument('name');

        //$name = trim($name);
        //获取配置
        $defail = $this->getConfigDefail($name);

        //创建一个资源连接
        $connection = $this->connectRabbitMQ();


        $defail['queue_name'] = $defail['queue_name'].'_'.env('domain.app_id');
        $defail['exchange_name'] = $defail['exchange_name'].'_'.env('domain.app_id');
       // dump($defail);exit;
        //建立通道
        $channel = $connection->channel();

        //控制流量
       // $channel->basic_qos(null, 1, null);

        //初始化交换机
        $channel->exchange_declare($defail['exchange_name'], $defail['exchange_type'], false, true, false);

        //初始化队列
        $channel->queue_declare($defail['queue_name'], false, true, false, false);

        //绑定队列与交换机
        $channel->queue_bind($defail['queue_name'], $defail['exchange_name'], $defail['route_key']);
        //定义一个处理函数
        $callback = function($msg) use ($output,$channel){
            //$output->write($msg->body);
            $body = $msg->body;
            $bodyArr = json_decode($body,true);
            $dao = app()->make(AdminApiLogDao::class);
            $bodyArr['delivery_tag'] = $msg->delivery_info['delivery_tag'];
            try {
                if($dao->createData('login',$bodyArr)){
                    $output->write('ok');
                    $channel->basic_ack($msg->delivery_info['delivery_tag']);
                }else{
                    $output->write('error');
                };

            }catch (\Exception $exception){
                $output->write($exception->getMessage());
            }

        };
//        //声明队列，这个函数的 passive ，如果有不返回这个队列的信息，如果没有就抛出异常
//        $channel->queue_declare("hello",false,false,false,false,false);
        //从队列中消费数据
        /**
         * 参数说明
         * params queue 队列名称
         * params consumer_tag 消费者标签
         * params no_local AMQP标准  RabbitMQ没有实现， 默认为false
         * params no_ack 收到消息后是否不需要回复 如果为true表示不回复， 一般是要回复的 设置为 false
         * params exclusive 排他消费，即这个队列只能由一个消费者来消费
         * params nowait 相当于一个异步，在执行完成之后返回结果，不堵塞
         * params callback 处理消息的回调方法
         */
        $channel->basic_consume($defail['queue_name'], $defail['consumer_tag'],false,false,false,false,$callback);
        //退出
        register_shutdown_function([$this, 'shutdown'], $channel, $connection);
        //等待队列执行完成
        while(count($channel->callbacks)){
            $channel->wait();
        }
    }


}
